package com.atguigu.Flink.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//* 算子链 :  将一个作业中多个task合并成一个大的task,就形成了算子链 。
//        *
//        * Flink数据的分发规则(分区规则): ChannelSelector
//        *    RebalancePartitioner:  绝对负载均衡。    rebalance() ，如果上游和下游的并行度不一致， 默认的分发规则就是rebalance。
//        *    RescalePartitioner:    相对负载均衡。    rescale()
//        *    KeyGroupStreamPartitioner： 按key分区   keyBy()
//        *    GlobalPartitioner：    强制并行度为1     global()
//        *    ShufflePartitioner：   随机发送。        shuffle()
//        *    BroadcastPartitioner： 广播。           broadcast()
//        *    ForwardPartitioner：   直连            forward()
//        *    CustomPartitionerWrapper
//        *
//        *
//        * 合并的条件:
//        *    上下游的算子的并行度一样， 且数据的分发规则为forward 。
//        *
//        * 为什么要合并：
//        *   1. 节省资源。
//        *   2. 可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。
//        *
//        *
//        * 算子链的操作:
//        *   Flink默认情况下会自动合并算子链。
//        *  基于算子设置:
//        *    startNewChain() : 开启新的算子链。
//        *    disableChaining() : 禁用算子链
//        *  全局禁用算子链:
//        *    env.disableOperatorChaining() ;

public class Flink07_OperatorChain {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port" , 6789);
        // 1. 准备流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //设置全局并行度
        env.setParallelism(1) ;

        //全局禁用算子链
        env.disableOperatorChaining() ;

        // 2. 从数据源读取数据
        // DataStreamSource => DataStream
        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888) ;

        //DataStream<String> rebalance = ds.rebalance();
        //DataStream<String> rescale = ds.rescale();
        //DataStream<String> global = ds.global();
        //DataStream<String> shuffle = ds.shuffle();
        //DataStream<String> broadcast = ds.broadcast();
        //DataStream<String> forward = ds.forward();

        // 3. 转换处理
        // 3.1 将读取到的一行数据按照分隔符进行切分，处理成（word, 1）的格式
        // SingleOutputStreamOperator => DataStream
        SingleOutputStreamOperator<Tuple2<String, Long>> flatMap = ds.flatMap(
                new FlatMapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1L));
                        }
                    }
                }
        );

        SingleOutputStreamOperator<Tuple2<String, Long>> map = flatMap
                .map(x -> x).returns(Types.TUPLE(Types.STRING , Types.LONG)).name("map1")
                //.startNewChain()
                //.disableChaining()
                .map(x -> x).returns(Types.TUPLE(Types.STRING , Types.LONG)).name("map2")
                .map(x -> x).returns(Types.TUPLE(Types.STRING , Types.LONG)).name("map3")
                .map(x -> x).returns(Types.TUPLE(Types.STRING , Types.LONG)).name("map4");


        //3.2 根据单词分组
        //KeyedStream => DataStream
        KeyedStream<Tuple2<String, Long>, String> keyBy = map.keyBy(
                new KeySelector<Tuple2<String, Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f0; // 使用Tuple的第一个元素作为Key
                    }
                }
        );
        //3.3 统计每个单词出现的次数
        /*
           sum(int) : 如果流中的数据类型是Tuple, 指定Tuple中的第几个元素用于sum汇总。
           sum(String) : 如果流中的数据类型是POJO ， 指定POJO中的哪个属性用于sum汇总。
         */
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyBy.sum(1);

        //4. 输出结果
        sum.print() ;

        //5. 启动执行
        env.execute();

    }
}
